#!/usr/bin/env python
# -*- coding: UTF-8 -*-
'''
@Project ：henan_sumo
@File ：main_send.py
@IDE ：PyCharm
@Author ：xinyingjie
@Date ：2025/4/16 17:48
'''
import threading
from copy import copy
from multiprocessing import Queue, Process
from loguru import logger
from datetime import datetime, timedelta
import time
import json
from kafka import KafkaConsumer, KafkaProducer
from config.config import KAFKA_HOST

logger.add("/sumo/main_send_kako_hsy.log", format="{time} {level} {message}", level="DEBUG")


# CJ：出京 、 JJ：进京

class DataFusion:
    def __init__(self, ):
        # 雷达数据保留多长时间(分钟)
        self.leida_data_max_time = 1000 * 60 * 3
        # 缓存保留时间(毫秒)
        self.cache_max_time = 1000 * 60 * 60
        # 首个卡口的数据
        self.first_sn_list = [11, 14, 19, 20, 24]
        # 卡口数据上报topic
        self.send_kako_topic_CJ = 'send_kako_topic_CJ'
        self.send_kako_topic_JJ = 'send_kako_topic_JJ'
        # 配置文件中sn集合
        self.sn_list_CJ = [11, 12, 13, 14, 15, 16, 17, 18]
        self.sn_list_JJ = [19, 20, 21, 22, 23, 24, 25, 26]
        self.sn_list = self.sn_list_CJ + self.sn_list_JJ
        self.sn_point_dist = {
            11: {'sn': 11, 'lon': 116.4236011, 'lat': 39.77833934},
            12: {'sn': 12, 'lon': 116.4233678, 'lat': 39.77179096},
            15: {'sn': 15, 'lon': 116.4261857, 'lat': 39.75540889},
            16: {'sn': 16, 'lon': 116.4285662, 'lat': 39.7464115},
            18: {'sn': 18, 'lon': 116.4306522, 'lat': 39.73931893},
            19: {'sn': 19, 'lon': 116.4310726, 'lat': 39.73943829},
            21: {'sn': 21, 'lon': 116.4280638, 'lat': 39.75005531},
            22: {'sn': 22, 'lon': 116.4256975, 'lat': 39.75950848},
            25: {'sn': 25, 'lon': 116.4239826, 'lat': 39.77116232},
            26: {'sn': 26, 'lon': 116.4240484, 'lat': 39.77798255},
            13: {'sn': 13, 'lon': 116.42371, 'lat': 39.76847134},
            14: {'sn': 14, 'lon': 116.4247321, 'lat': 39.76152304},
            17: {'sn': 17, 'lon': 116.4290889, 'lat': 39.74394431},
            20: {'sn': 20, 'lon': 116.4299701, 'lat': 39.7434342},
            23: {'sn': 23, 'lon': 116.4254769, 'lat': 39.7616087},
            24: {'sn': 24, 'lon': 116.4244934, 'lat': 39.7685285}
        }
        # 接受卡口数据的topic
        self.receive_kako_topic_list = ['kako_data_CJ']
        # 接受雷达数据的topic
        self.receive_leida_topic_list = ['CJ-10-SUM']
        # 雷达缓存数据 {key: [(timestamp1, value1), (timestamp2, value2)]}
        self.leida_data_cache_dist = {}
        # 相机中 station 对应的车道总数
        self.station_totalLaneNum_dist = {11: 5, 12: 5, 13: 2, 14: 2, 15: 4, 16: 4, 17: 2, 18: 4, 19: 4, 20: 2, 21: 4,
                                          22: 4, 23: 2, 24: 2, 25: 4, 26: 4}
        # 车牌与首个车辆信息对应关系
        self.plate_carInfo_dist = {}
        # 车牌与全域id对应关系
        self.plate_globalId_dist = {}
        self.kafka_producer = KafkaProducer(bootstrap_servers=KAFKA_HOST, key_serializer=str.encode,
                                            value_serializer=lambda x: json.dumps(x).encode('utf-8'))

    # 启动线程消费并生成数据
    def consumer_producer_kafka(self):
        # 删除缓存数据
        threading.Thread(target=self.del_cache_data, ).start()
        # 拉取雷达数据
        if len(self.receive_leida_topic_list) > 0:
            for receive_kako_topic in self.receive_leida_topic_list:
                threading.Thread(target=self.receive_leida_data, args=(receive_kako_topic,)).start()
        # 拉取卡口数据
        if len(self.receive_kako_topic_list) > 0:
            for receive_kako_topic in self.receive_kako_topic_list:
                threading.Thread(target=self.receive_kako_data, args=(receive_kako_topic,)).start()

    # 删除缓存数据
    def del_cache_data(self):
        while True:
            try:
                # logger.info(f'当前雷达感知数据 = {self.leida_data_cache_dist}')
                leida_data_cache_dist = self.leida_data_cache_dist.copy()
                leida_data_cache_dist_new = {}
                if len(leida_data_cache_dist) > 0:
                    for k, v in leida_data_cache_dist.items():
                        cun_time = int(time.time() * 1000)
                        if len(v) > 0:
                            new_list = []
                            for i in v:
                                if cun_time - i['timestamp'] < self.leida_data_max_time:
                                    new_list.append(i)
                            leida_data_cache_dist_new[k] = new_list
                    self.leida_data_cache_dist = leida_data_cache_dist_new

                if len(self.plate_carInfo_dist) > 0:
                    cun_time = int(time.time() * 1000)
                    plate_carInfo_dist = self.plate_carInfo_dist.copy()
                    for plate, carInfo in plate_carInfo_dist.items():
                        timestamp = carInfo['timestamp']
                        if cun_time - timestamp > self.cache_max_time:
                            logger.info(f"【del_cache_data】删除缓存数据：{plate}")
                            self.plate_carInfo_dist.pop(plate)
                            self.plate_globalId_dist.pop(plate)
            except Exception as e:
                logger.error(f"删除缓存数据异常{e}")
            time.sleep(60 * 1)

    # 消费卡口数据、匹配、发送
    def receive_kako_data(self, kafka_topic):
        kafka_consumer_e1 = KafkaConsumer(
            kafka_topic,
            bootstrap_servers=KAFKA_HOST,
            auto_offset_reset='latest',
            enable_auto_commit=False
        )

        while True:
            time.sleep(0.1)
            for message in kafka_consumer_e1:
                if message:
                    try:
                        message_value = message.value.decode('utf-8')
                        kako_info = json.loads(message_value)
                        station = kako_info['stationId']
                        car_id = kako_info['vehicleId']
                        kako_laneNum = kako_info['laneNum']
                        global_id = str(kako_info['vehicleId']) + "-" + str(station) + "-" + str(
                            int(time.time() * 1000))
                        if station not in self.sn_list:
                            continue
                        target_info = {'id': car_id, 'station': station,
                                       'lon': self.sn_point_dist[station]['lon'],
                                       'lat': self.sn_point_dist[station]['lat'], 'angle': 0,
                                       'carType': 1, 'picLicense': kako_info['picLicense'],
                                       'lane': kako_laneNum, 'speed': 80,
                                       'orgCode': '', 'timestamp': int(time.time() * 1000),
                                       'global_id': global_id}

                        # 判断当前数据的sn是否在已有的配置文件中
                        if station in self.sn_list:
                            if station in self.first_sn_list:
                                picLicense = kako_info['picLicense']
                                # 获取雷达缓存数据，计算速度
                                leida_data_cache_dist = self.leida_data_cache_dist.copy()
                                # 过滤车道号
                                leida_data_list = leida_data_cache_dist.get(station)
                                leida_data_list_new = []
                                if leida_data_list is not None and len(leida_data_list) > 0:
                                    for leida_data in leida_data_list:
                                        if leida_data['lane'] == kako_laneNum:
                                            leida_data_list_new.append(leida_data)
                                # 转换车道号
                                if station in self.station_totalLaneNum_dist.keys():
                                    totalLaneNum = self.station_totalLaneNum_dist[station]
                                    target_info['lane'] = totalLaneNum - target_info['lane']
                                else:
                                    continue
                                if leida_data_list_new is not None and len(leida_data_list_new) > 0:
                                    # 提取所有speed值
                                    speeds = [item['speed'] for item in leida_data_list_new]
                                    # 计算平均值
                                    average_speed = sum(speeds) / len(speeds) if speeds is not None else 80
                                    target_info['speed'] = round(average_speed, 1)
                                    carType = leida_data_list_new[0]['carType']
                                    # 设置车型 TODO
                                    target_info['carType'] = carType
                                # 缓存数据
                                self.plate_carInfo_dist[picLicense] = target_info
                                self.plate_globalId_dist[picLicense] = global_id
                                # 上报数据
                                if station in self.sn_list_CJ:
                                    logger.info(f"出京卡口上报{target_info}")
                                    send_topic = self.send_kako_topic_CJ
                                else:
                                    logger.info(f"入京卡口上报{target_info}")
                                    send_topic = self.send_kako_topic_JJ
                                self.kafka_producer.send(send_topic, value=target_info,
                                                         key=str(int(time.time() * 1000)))
                            else:
                                pic_license = target_info['picLicense']
                                if '默A' in pic_license:
                                    continue
                                plate_globalId_dist = self.plate_globalId_dist.copy()
                                if pic_license not in plate_globalId_dist.keys():
                                    continue
                                else:
                                    global_id = plate_globalId_dist[pic_license]
                                    # 设置车型（根据缓存设置）
                                    plate_carInfo_dist = self.plate_carInfo_dist.copy()
                                    if pic_license not in plate_carInfo_dist.keys():
                                        continue
                                    car_info = plate_carInfo_dist[pic_license]
                                    target_info['carType'] = car_info['carType']
                                    # 获取雷达缓存数据，计算速度
                                    leida_data_cache_dist = self.leida_data_cache_dist.copy()
                                    # 过滤车道号
                                    leida_data_list = leida_data_cache_dist.get(station)
                                    leida_data_list_new = []
                                    if leida_data_list is not None and len(leida_data_list) > 0:
                                        for leida_data in leida_data_list:
                                            if leida_data['lane'] == kako_laneNum:
                                                leida_data_list_new.append(leida_data)
                                    if len(leida_data_list_new) > 0:
                                        # 提取所有speed值
                                        speeds = [item['speed'] for item in leida_data_list_new]
                                        # 计算平均值
                                        average_speed = sum(speeds) / len(speeds) if speeds is not None else 80
                                        target_info['speed'] = round(average_speed, 1)
                                    # 转换车道号
                                    if station in self.station_totalLaneNum_dist.keys():
                                        totalLaneNum = self.station_totalLaneNum_dist[station]
                                        target_info['lane'] = totalLaneNum - target_info['lane']
                                        target_info['timestamp'] = int(time.time() * 1000)
                                        target_info['global_id'] = global_id
                                        # 上报数据
                                        if station in self.sn_list_CJ:
                                            logger.info(f"出京卡口上报{target_info}")
                                            send_topic = self.send_kako_topic_CJ
                                        else:
                                            logger.info(f"入京卡口上报{target_info}")
                                            send_topic = self.send_kako_topic_JJ
                                        self.kafka_producer.send(send_topic, value=target_info,
                                                                 key=str(int(time.time() * 1000)))
                    except Exception as e:
                        logger.error(f"【相机数据 = {kafka_topic}】kafka数据处理异常{e}")

    # 消费雷达kafka数据
    def receive_leida_data(self, kafka_topic):
        kafka_consumer_e1 = KafkaConsumer(
            kafka_topic,
            bootstrap_servers=KAFKA_HOST,
            auto_offset_reset='latest',
            enable_auto_commit=False
        )
        while True:
            time.sleep(0.1)
            for message in kafka_consumer_e1:
                if message:
                    try:
                        message_value = message.value.decode('utf-8')
                        info = json.loads(message_value)
                        if 'targetList' not in info.keys():
                            continue
                        targetList = info['targetList']
                        if len(targetList) > 0:
                            for target in targetList:
                                station = target['station']
                                if 'lon' not in target.keys() or 'speed' not in target.keys() or 'carType' not in target.keys() or 'lane' not in target.keys() or \
                                        target['lon'] is None or target['speed'] is None or target['carType'] is None or \
                                        target['lane'] is None or int(target['lon']) == 0 or int(target['speed']) == 0:
                                    continue
                                # 添加新数据 (时间戳, 值) 到列表
                                if station not in self.leida_data_cache_dist:
                                    self.leida_data_cache_dist[station] = []
                                target['timestamp'] = int(time.time() * 1000)
                                self.leida_data_cache_dist[station].append(target)

                    except Exception as e:
                        logger.error(f"【感知数据 = {kafka_topic}】kafka数据处理异常{e}")

        # 移除该 key 下超过 3 分钟的数据


if __name__ == '__main__':
    data_fusion = DataFusion()
    data_fusion.consumer_producer_kafka()
